package cn.zwx.business.order.test.simple;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.Charset;
import java.util.List;
import java.util.jar.JarEntry;

/**
 * @description: 普通消息
 * @projectName:rocket-mq-example
 * @see:cn.zwx.business.order.test.simple
 * @author:zhangwenxue
 * @createTime:2020/10/16 12:04
 * @version:1.0
 */
public class MeaggessApp {

    /**
     * 生产者组
     */
    private static String PRODUCE_RGROUP = "test_producer";
    /**
     * 创建生产者对象
     */
    private static DefaultMQProducer producer = null;

    static {
        producer = new DefaultMQProducer(PRODUCE_RGROUP);
        //不开启vip通道 开通口端口会减2
        producer.setVipChannelEnabled(false);
        //绑定name server
//        producer.setNamesrvAddr("192.168.127.128:9876");
        producer.setNamesrvAddr("localhost:9876");
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }
    /**
     * 性能对比
     * 发送方式	  发送TPS	发送结果反馈	  可靠性
     * 同步发送    快	       有	      不丢失
     * 异步发送	  快	      有	      不丢失
     * 单向发送	  最快       无	        可能丢失
     **/
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        long startTime = System.currentTimeMillis();
        send_syn();//807ms
//        send_asyn();//1ms+回调时间
//        one_way();//811ms
          consumer();
        long endTime = System.currentTimeMillis();
        System.out.println("totalTime:"+(endTime-startTime));
    }

    /**
     * @description 消费者
     * @author zhangwenxue
     * @createTime 2020/10/16 12:16
     **/
    private static void consumer() {
        //实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup1");

         /**
            * PullRequest这里说明一下，上面我们已经提了一下rocketmq的push模式其实是通过pull模式封装实现的，pullrequest这里是通过长轮询的方式达到push效果。
            * 长轮询方式既有pull的优点又有push模式的实时性有点。
            push方式是server端接收到消息后，主动把消息推送给client端，实时性高。弊端是server端工作量大，影响性能，其次是client端处理能力不同且client
            端的状态不受server端的控制，如果client端不能及时处理消息容易导致消息堆积已经影响正常业务等。
            pull方式是client循环从server端拉取消息，主动权在client端，自己处理完一个消息再去拉取下一个，缺点是循环的时间不好设定，时间太短容易忙等，
            浪费CPU资源，时间间隔太长client的处理能力会下降，有时候有些消息会处理不及时。
         长轮询的方式可以结合两者优点
            1、检查PullRequest对象中的ProcessQueue对象的dropped是否为true（在RebalanceService线程中为topic下的MessageQueue创建拉取消息请求时要维护对应的ProcessQueue对象，若Consumer不再订阅该topic则会将该对象的dropped置为true）；若是则认为该请求是已经取消的，则直接跳出该方法；
            2、更新PullRequest对象中的ProcessQueue对象的时间戳（ProcessQueue.lastPullTimestamp）为当前时间戳；
            3、检查该Consumer是否运行中，即DefaultMQPushConsumerImpl.serviceState是否为RUNNING;若不是运行状态或者是暂停状态（DefaultMQPushConsumerImpl.pause=true），
               则调用PullMessageService.executePullRequestLater(PullRequest pullRequest, long timeDelay)方法延迟再拉取消息，其中timeDelay=3000；该方法的目的是在3秒之后再次将该PullRequest对象放入PullMessageService. pullRequestQueue队列中；并跳出该方法；
            4、进行流控。若ProcessQueue对象的msgCount大于了消费端的流控阈值（DefaultMQPushConsumer.pullThresholdForQueue，默认值为1000），
               则调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法；
            5、若不是顺序消费（即DefaultMQPushConsumerImpl.consumeOrderly等于false），则检查ProcessQueue对象的msgTreeMap:TreeMap<Long,MessageExt>变量的第一个key值与最后一个key值之间的差额，
               该key值表示查询的队列偏移量queueoffset；若差额大于阈值（由DefaultMQPushConsumer. consumeConcurrentlyMaxSpan指定，默认是2000），则调用PullMessageService.executePullRequestLater方法，在50毫秒之后重新将该PullRequest请求放入PullMessageService.pullRequestQueue队列中；并跳出该方法；
            6、以PullRequest.messageQueue对象的topic值为参数从RebalanceImpl.subscriptionInner: ConcurrentHashMap, SubscriptionData>中获取对应的SubscriptionData对象，若该对象为null，考虑到并发的关系，调用executePullRequestLater方法，稍后重试；并跳出该方法；
            7、若消息模型为集群模式（RebalanceImpl.messageModel等于CLUSTERING），则以PullRequest对象的MessageQueue变量值、type =READ_FROM_MEMORY（从内存中获取消费进度offset值）
                为参数调用DefaultMQPushConsumerImpl. offsetStore对象（初始化为RemoteBrokerOffsetStore对象）的readOffset(MessageQueue mq, ReadOffsetType type)方法从本地内存中获取消费进度offset值。若该offset值大于0 则置临时变量commitOffsetEnable等于true否则为false；
               该 offset值作为pullKernelImpl方法中的commitOffset参数，在Broker端拉取消息之后根据commitOffsetEnable参数值决定是否用该offset更新消息进度。该readOffset方法的逻辑是：以入参MessageQueue对象从RemoteBrokerOffsetStore.offsetTable:ConcurrentHashMap <MessageQueue,AtomicLong>变量中获取消费进度偏移量；若该偏移量不为null则返回该值，否则返回-1；
            8、当每次拉取消息之后需要更新订阅关系（由DefaultMQPushConsumer. postSubscriptionWhenPull参数表示，默认为false）并且以topic值参数从RebalanceImpl.subscriptionInner获取的SubscriptionData对象的classFilterMode等于false（默认为false），则将sysFlag标记的第3个字节置为1，否则该字节置为0；
            9、该sysFlag标记的第1个字节置为commitOffsetEnable的值；第2个字节（suspend标记）置为1；第4个字节置为classFilterMode的值；
            10、 初始化匿名内部类PullCallback，实现了onSucess/onException方法； 该方法只有在异步请求的情况下才会回调；
            11、调用底层的拉取消息API接口：PullAPIWrapper.pullKernelImpl(MessageQueue mq, String subExpression, long subVersion,long offset, int maxNums, int sysFlag,long commitOffset,long brokerSuspendMaxTimeMillis, long timeoutMillis, CommunicationMode communicationMode, PullCallback pullCallback)方法进行消息拉取操作。将回调类PullCallback传入该方法中，当采用异步方式拉取消息时，在收到响应之后会回调该回调类的方法。
        **/
//        DefaultLitePullConsumer consumerGroup1 = new DefaultLitePullConsumer("consumerGroup1");
        consumer.setNamesrvAddr("localhost:9876");
        try {
            //订阅topic1下的所有tag的消息
            consumer.subscribe("topic_family", "*");
            //注册回调
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
//                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), JSON.toJSONString(msgs.get(0)));
                    System.out.println("接到消息:"+new String(msgs.get(0).getBody(), Charset.defaultCharset()));
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            System.out.printf("Consumer Started.%n");

        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    /**
     * @description 单项发送
     * 单向（Oneway）发送特点为发送方只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。
     * 此方式发送消息的过程耗时非常短，一般在微秒级别
     * 场景:适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。
     **/
    private static void one_way() throws InterruptedException, RemotingException, MQClientException {
        //创建消息
        Message message = new Message("topic_family", (" 单项发送 ").getBytes());
        //同步发送消息
        producer.sendOneway(message);
        System.out.println("Product:单项发送成功!!!");
    }


    /**
     * @description 异步发送
     * 异步发送是指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。
     * 消息队列 RocketMQ 的异步发送，需要用户实现异步发送回调接口（SendCallback）
     * 场景:异步发送一般用于链路耗时较长，对 RT 响应时间较为敏感的业务场景，例如批量发货等操作。
     **/
    private static void send_asyn() throws RemotingException, MQClientException, InterruptedException {
        //创建消息
        Message message = new Message("topic_family", ("  异步发送  ").getBytes());
        //异步发送消息
        producer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("Product-异步发送-输出信息={}"+ sendResult);
            }
            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                //补偿机制，根据业务情况进行使用，看是否进行重试
            }
        });
    }

    /**
     * @description 同步发送
     * 同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
     * 场景:此种方式应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等。
     **/
    private static void send_syn() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        //创建消息
        Message message = new Message("topic_family", ("  同步发送  ").getBytes());
        //设置延时消息
//        1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        message.setDelayTimeLevel(3);
        //同步发送消息
//       FLUSH_DISK_TIMEOUT
//       如果Broker设置MessageStoreConfig的FlushDiskType = SYNC_FLUSH（默认为ASYNC_FLUSH），并且Broker没有在MessageStoreConfig的syncFlushTimeout（默认为5秒）内完成刷新磁盘，您将获得此状态。
//       FLUSH_SLAVE_TIMEOUT
//       如果Broker的角色是SYNC_MASTER（默认为ASYNC_MASTER），并且从属Broker未在MessageStoreConfig的syncFlushTimeout（默认为5秒）内完成与主服务器的同步，则您将获得此状态。
//       SLAVE_NOT_AVAILABLE
//       如果Broker的角色是SYNC_MASTER（默认为ASYNC_MASTER），但没有配置slave Broker，您将获得此状态。
//       SEND_OK
//       SEND_OK并不意味着它是可靠的。要确保不会丢失任何消息，还应启用SYNC_MASTER或SYNC_FLUSH。
        SendResult sendResult = producer.send(message);

        System.out.println("Product-同步发送-Product信息={}"+sendResult);
    }
}
